#include "../common/config.h"
#include <mysql_connection.h>
#include <cppconn/driver.h>
#include <cppconn/exception.h>
#include <cppconn/resultset.h>
#include <cppconn/statement.h>
#include <cppconn/prepared_statement.h>
#include "../common/dbs.h"



LoggerPtr gLogger = Logger::getLogger("fsmonitor");
int  gRunThread = 1;

struct call_state
{
    string uuid;
    string acct;
    string caller;
    string called;
    CALL_TYPE calltype;
    string create_time;
    string ring_time;
    string answer_time;
    string host;
    string ip;
    string state;
    int    hangup;
};

apr_queue_t* conn_queue = NULL;
apr_queue_t* update_queue = NULL;

map<string, apr_queue_t*> monitor_queue;

void * APR_THREAD_FUNC db_hangup(apr_thread_t *thd, void *data)
{
retry:
    sql::Connection *con = GetDBC();

    try
    {
        auto_ptr<sql::PreparedStatement> stmt;
        auto_ptr<sql::ResultSet> res;

        if (con != NULL)
        {
            LOG4CXX_INFO(gLogger, "get database connect  success");
        }
        else
        {
            LOG4CXX_WARN(gLogger, "get database connect  failed");
        }

        stmt.reset(con->prepareStatement("select uuid,host from call_status where call_status.hangup =1 order by id asc"));

        while (gRunThread)
        {
            res.reset(stmt->executeQuery());

            while (res->next())
            {
                string* strUuid = new string(res->getString(1));
                string strHost = res->getString(2);
                apr_queue_push(monitor_queue[strHost], strUuid);
            }

            apr_sleep(100000);
        }

        ReleaseDBC(con);
    }
    catch (sql::SQLException &e)
    {
        LOG4CXX_WARN(gLogger, "err:" << e.what() << " (MySQL error code: " << e.getErrorCode() << ", SQLState: " << e.getSQLState());
        goto retry;
    }

    return 0;
}
void * APR_THREAD_FUNC update_status(apr_thread_t *thd, void *data)
{
retry:
    sql::Connection *con = GetDBC();

    try
    {
        auto_ptr<sql::PreparedStatement> stmt_insert, stmt_update, stmt_delete;
        auto_ptr<sql::ResultSet> res;

        if (con != NULL)
        {
            LOG4CXX_INFO(gLogger, "get database connect  success");
        }
        else
        {
            LOG4CXX_WARN(gLogger, "get database connect  failed");
        }

        stmt_insert.reset(con->prepareStatement("insert into call_status(uuid,host,create_time,state) VALUES(?,?,?,?)"));
        stmt_update.reset(con->prepareStatement("update call_status set calltype=?,acct=?,caller=?,called=?,ip=?,ring_time=from_unixtime(?),answer_time=from_unixtime(?),state=? where uuid=?"));
        stmt_delete.reset(con->prepareStatement("delete from call_status where uuid=?"));

        while (gRunThread)
        {
            call_state* pState = NULL;

            while (apr_queue_pop(update_queue, (void**)&pState) == APR_SUCCESS)
            {
                if (!pState->create_time.empty())
                {
                    stmt_insert->setString(1, pState->uuid);
                    stmt_insert->setString(2, pState->host);
                    stmt_insert->setDateTime(3, pState->create_time);
                    stmt_insert->setString(4, pState->state);
                    res.reset(stmt_insert->executeQuery());

                }
                else if (pState->hangup == 1)
                {
                    stmt_delete->setString(1, pState->uuid);
                    res.reset(stmt_delete->executeQuery());
                }
                else
                {
                    stmt_update->setInt(1, pState->calltype);
                    stmt_update->setString(2, pState->acct);
                    stmt_update->setString(3, pState->caller);
                    stmt_update->setString(4, pState->called);
                    stmt_update->setString(5, pState->ip);
                    stmt_update->setString(6, pState->ring_time);
                    stmt_update->setString(7, pState->answer_time);
                    stmt_update->setString(8, pState->state);
                    stmt_update->setString(9, pState->uuid);
                    res.reset(stmt_update->executeQuery());
                }

                delete pState;
            }
        }

        ReleaseDBC(con);
    }
    catch (sql::SQLException &e)
    {
        LOG4CXX_WARN(gLogger, "err:" << e.what() << " (MySQL error code: " << e.getErrorCode() << ", SQLState: " << e.getSQLState());
        goto retry;
    }

    return 0;
}
struct callserver_info_t
{
    char* server;
    int  port;
    char* user;
    char* pwd;
    apr_queue_t* hangup_queue;
    callserver_info_t()
            : server(NULL)
            , port(8021)
            , user(NULL)
            , pwd(NULL)
    {

    }
    ~callserver_info_t()
    {
        if (server != NULL)
        {
            free(server);
            server = NULL;
        }

        if (user != NULL)
        {
            free(user);
            user = NULL;
        }

        if (pwd != NULL)
        {
            free(pwd);
            pwd = NULL;
        }
    }
};

void event_update_status(esl_event_t* ievent, const char* pHost)
{
    const char* uuid = esl_event_get_header(ievent, "unique-id");
    const char* pcalltype = esl_event_get_header(ievent, "variable_calltype");

    if (!uuid)
    {
        return;
    }

    call_state* pState = NULL;
    char *pacct, *pcaller, *pcalled;
    const char* stat = esl_event_get_header(ievent, "Answer-State");

	if (!strcasecmp(esl_event_get_header(ievent, "event-name"), "CHANNEL_CREATE"))
	{
		pState = new call_state();
		const char* create_time = esl_event_get_header(ievent, "Event-Date-Local");
		pState->create_time = create_time;
		pState->uuid = uuid;
		pState->host = pHost;
		pState->state = stat;
		apr_queue_push(update_queue, pState);
		return;
	}

    if (!pcalltype)
    {
        pcalltype = "-1";
        pacct = esl_event_get_header(ievent, "caller-caller-id-number");
        pcaller = esl_event_get_header(ievent, "caller-caller-id-number");
        pcalled = esl_event_get_header(ievent, "Caller-Destination-Number");
    }
    else
    {
        pacct = esl_event_get_header(ievent, "variable_orig_acct");
        pcaller = esl_event_get_header(ievent, "variable_orig_caller");
        pcalled = esl_event_get_header(ievent, "variable_orig_called");
    }

    CALL_TYPE call_type = (CALL_TYPE) atoi(pcalltype);

    const char* pTohost = esl_event_get_header(ievent, "variable_sip_to_host");

    const char* ring_time = esl_event_get_header(ievent, "Caller-Channel-Progress-Media-Time");

    const char* answer_time = esl_event_get_header(ievent, "Caller-Channel-Answered-Time");
    pState = new call_state();
    pState->uuid = uuid;
    pState->acct = pacct;
    pState->caller = pcaller;
    pState->called = pcalled;
    pState->calltype = call_type;
    pState->ring_time = ring_time;

    if (pState->ring_time.length() > 6)
    {
        pState->ring_time = pState->ring_time.substr(0, pState->ring_time.length() - 6);
    }

    pState->answer_time = answer_time;

    if (pState->answer_time.length() > 6)
    {
        pState->answer_time = pState->answer_time.substr(0, pState->answer_time.length() - 6);
    }

    pState->host = pHost;
    pState->ip = pTohost;
    pState->state = stat;
    pState->hangup = 0;
    apr_queue_push(update_queue, pState);
}
void event_hangup(esl_event_t* ievent, const char* pHost)
{
    const char* uuid = esl_event_get_header(ievent, "unique-id");

    if (!uuid)
    {
        return;
    }

    call_state* pState = new call_state();
    pState->uuid = uuid;
    pState->hangup = 1;
    apr_queue_push(update_queue, pState);
}


static void *msg_thread_run(esl_thread_t *me, void *obj)
{
    callserver_info_t*   pCtrl = (callserver_info_t*) obj;

    esl_handle_t handle = {{0}};

    if (esl_connect(&handle, pCtrl->server, pCtrl->port, pCtrl->user, pCtrl->pwd) == ESL_SUCCESS)
    {
        LOG4CXX_INFO(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " success");
    }
    else
    {
        LOG4CXX_WARN(gLogger, "connect " << pCtrl->server << ":" << pCtrl->port << " failed");
        goto done;
    }

    if (esl_events(&handle, ESL_EVENT_TYPE_PLAIN, "CHANNEL_CREATE  CHANNEL_PROGRESS_MEDIA CHANNEL_BRIDGE CHANNEL_ANSWER CHANNEL_HANGUP") == ESL_FAIL)
    {
        LOG4CXX_WARN(gLogger, pCtrl->server << " set event plain faild,contact admin");
        goto done;
    }

    while (handle.connected)
    {
        string* puuid = NULL;

        while (apr_queue_trypop(pCtrl->hangup_queue, (void**) &puuid) == APR_SUCCESS)
        {
            esl_execute(&handle, "hangup", NULL, puuid->c_str());
            delete puuid;
            puuid = NULL;
        }

        esl_status_t status = esl_recv_event_timed(&handle, 10, 1, NULL);

        if (status == ESL_FAIL)
        {
            LOG4CXX_WARN(gLogger, pCtrl->server << " Disconnected.wait for reconnect! ");
            goto done;
        }
        else if (status == ESL_SUCCESS)
        {
            if (handle.last_event)
            {
                const char *type = esl_event_get_header(handle.last_event, "content-type");

                if (!esl_strlen_zero(type))
                {
                    if (!strcasecmp(type, "text/event-plain"))
                    {
                        const char* eventname = esl_event_get_header(handle.last_ievent, "event-name");

                        if (!esl_strlen_zero(eventname))
                        {
                            if ((!strcasecmp(eventname, "CHANNEL_CREATE"))
                                    || !strcasecmp(eventname, "CHANNEL_PROGRESS_MEDIA")
                                    || (!strcasecmp(eventname, "CHANNEL_BRIDGE"))
                                    || (!strcasecmp(eventname, "CHANNEL_ANSWER")))
                            {
                                event_update_status(handle.last_ievent, pCtrl->server);
                            }
                            else if (!strcasecmp(eventname, "CHANNEL_HANGUP"))
                            {
                                event_hangup(handle.last_ievent, pCtrl->server);
                            }
                        }
                    }
                }
            }
        }

        apr_sleep(1000);

    }

done:
    apr_sleep(APR_USEC_PER_SEC);
    apr_queue_push(conn_queue, pCtrl);
    return NULL;
}
static void sighandler( int sig_no )
{
    gRunThread = 0;
}
#ifndef _DEBUG
int localMain(int argc, char* argv[]);
int main(int argc, char *argv[])
{
    if (init_daemon(argc, argv, localMain) < 0)
    {
        return 0;
    }

    return 0;
}
int localMain(int argc, char* argv[])
#else
int main(int argc, char *argv[])
#endif
{
    signal(SIGUSR1, sighandler);
    apr_initialize();
    CIniFile config;
    string strCfgPath("/etc/fsmonitor.ini");

    if (argc >= 2)
    {
        strCfgPath = argv[1];
    }

    ostringstream ss;

    config.SetPath(strCfgPath);
    config.ReadFile();
    PropertyConfigurator::configure(strCfgPath);

    const string password = config.GetValue("monitor", "fsstr", "66call");
    const string strClient = config.GetValue("monitor", "callservers", "");

    const int    nThreadPool = config.GetValueI("monitor", "threads", 100);
    const int    nQueuePool = config.GetValueI("monitor", "jobqueue", 1000);


    apr_thread_pool_t *thrp;


    //initalize线程池
    apr_pool_t *main_Pool;

    if (apr_pool_create(&main_Pool, NULL) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init memory pool ok");


    if (apr_queue_create(&conn_queue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init conn_queue queue pool ok");

    if (apr_queue_create(&update_queue, nQueuePool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_pool_create error");
        apr_queue_term(conn_queue);
        apr_pool_destroy(main_Pool);
        apr_terminate();
        return -1;
    }

    LOG4CXX_INFO(gLogger, "init update state queue pool ok");

    if (apr_thread_pool_create(&thrp, nThreadPool, nThreadPool, main_Pool) != APR_SUCCESS)
    {
        LOG4CXX_WARN(gLogger, "apr_thread_pool_create error,thread pool=" << nThreadPool);
        apr_queue_term(conn_queue);
        apr_queue_term(update_queue);
        apr_pool_destroy(main_Pool);
        apr_terminate();
        return -2;//线程池创建失败
    }

    if (!initDBS(config, "monitor-db", main_Pool))
    {
        LOG4CXX_WARN(gLogger, "init dbs error");
        apr_queue_term(conn_queue);
        apr_queue_term(update_queue);
        apr_pool_destroy(main_Pool);
        apr_terminate();
        return -3;//数据库初始化失败
    }

    //创建客户监视线程
    if (strClient.length() > 0)
    {
        string delimiter(",");
        const vector<string>& clientlist = split(strClient, delimiter);

        for (vector<string>::const_iterator iter = clientlist.begin(); iter != clientlist.end(); ++iter)
        {
            char* pIp = strdup((*iter).c_str());
            char* pPort = strchr(pIp, ':');

            if (pPort != NULL)
            {
                *pPort++ = '\0';
            }
            else
            {
                pPort = "8021";
            }

            callserver_info_t* pCtrl = new callserver_info_t();
            pCtrl->server = strdup(pIp);
            pCtrl->port = atoi(pPort);
            pCtrl->user = NULL;
            pCtrl->pwd = strdup(password.c_str());

            if (apr_queue_create(&pCtrl->hangup_queue, nQueuePool, main_Pool) != APR_SUCCESS)
            {
                delete pCtrl;
            }
            else
            {
                apr_queue_push(conn_queue, pCtrl);
                monitor_queue[string(pIp)] = pCtrl->hangup_queue;
            }

            free(pIp);
        }
    }

    esl_global_set_logger(NULL);
    //创建数据库监视线程
    apr_thread_pool_push(thrp, update_status, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);
    apr_thread_pool_push(thrp, db_hangup, NULL, APR_THREAD_TASK_PRIORITY_NORMAL, NULL);


    while (gRunThread)
    {
        callserver_info_t* pCtrl = NULL;
        apr_status_t ret = apr_queue_pop(conn_queue, (void**) & pCtrl);

        if (ret == APR_SUCCESS)
        {
            if (esl_thread_create_detached(msg_thread_run, pCtrl) != ESL_SUCCESS)
            {
                apr_queue_push(conn_queue, pCtrl);
            }
        }
        else if (ret == APR_EOF)
        {
            break;
        }
    }

    gRunThread = 0;
    apr_queue_term(conn_queue);
    apr_thread_pool_destroy(thrp);

    destoryDBS();
    apr_pool_destroy(main_Pool);
    apr_terminate();
    return 0;
}
